package com.example.lprxjava;

import android.util.Log;
import android.view.View;
import android.widget.TextView;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Function;

/**
 * Author by lp,on 2021/7/20/020,14:27.
 * Rxjava 组合
 */
public class MergeFragement extends BaseFragment {

    private static final String TAG = "MergeFragement";
    @Override
    String getTAG() {
        return TAG;
    }
    @Override
    String getTitle() {
        return "RxJava 组合操作符";
    }

    @Override
    int getFragmentLayoutId() {
        return R.layout.merge_frqagment;
    }

    @Override
    void onLoad(View view) {

        view.findViewById(R.id.btn_concat).setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                concat();
            }
        });
        view.findViewById(R.id.btn_concatArray).setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                concatArray();
            }
        });
        view.findViewById(R.id.btn_merge).setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                merge();
            }
        });


    }

    //concat 将观察者组合在一起，串行事件，最多4个  ,使用create的时候必须onComplete 否则不进行下一步
    //concatArray 与concat相同，但是没有个数限制
    //merge为并行执行事件,最多四个，mergeArray无限制
    private void concat() {
        observable = Observable.create(new ObservableOnSubscribe<Object>() {
            @Override
            public void subscribe(ObservableEmitter<Object> e) throws Exception {
                Log.i(TAG, "current Thread: " + Thread.currentThread().getName());
                e.onNext("concat方式创建...");
                e.onComplete();
            }
        });
        Observable observable1 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                Log.i(TAG, "current Thread: " + Thread.currentThread().getName());
                e.onNext("concat方式创建222...");
                e.onComplete();
            }
        });
        Observable<String> concat = Observable.concat(observable, observable1);
        concat.subscribe(observer);
    }

    private void concatArray() {
        Observable<String> objectObservable = Observable.concatArray(Observable.just("concatArray"), Observable.just("1111"), Observable.just("2222"));
        objectObservable.subscribe(observer);

    }

    private void merge() {
        Observable<String> merge = Observable.mergeArray(Observable.just("merge"), Observable.just("1111"), Observable.just("2222"), Observable.just("3333"), Observable.just("4444"));
        merge.safeSubscribe(observer);
    }


    //观察者
    private Observer observer = new Observer<Object>() {

        @Override
        public void onSubscribe(Disposable d) {
            Log.i(TAG, "onSubscribe: ");

        }

        @Override
        public void onNext(Object o) {
            Log.i(TAG, "current Thread: " + Thread.currentThread().getName());
            Log.i(TAG, "onNext: " + o);
            stringBuffer.append(o.toString() + "\r\n");
            showText();
        }

        @Override
        public void onError(Throwable e) {
            Log.i(TAG, "onError: " + e.getMessage());

        }

        @Override
        public void onComplete() {
            Log.i(TAG, "onComplete: ");
        }
    };


    @Override
    public void onDestroyView() {
        super.onDestroyView();
        Log.i(TAG, "onDestroyView: ");

    }
}
